package nl.altenpts.net.broker;

import java.io.IOException;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.util.LinkedList;

import nl.altenpts.net.ChannelHandler;
import nl.altenpts.net.NetReactor;

/**
 * Message broker that is able to send upd messages to different targets and
 * dispatch incoming upd messages.
 * 
 * @author fabian
 *
 */

public class MessageBroker {
	
	/**
	 * Create a new MessageBroker.
	 * 
	 * @param reactor A NetReactor which is used to send and receive udp messages.
	 * @param address as InetSocketAddress object which has a hostname and a socket
	 * @throws IOException
	 */
	public MessageBroker(NetReactor reactor, InetSocketAddress address) throws IOException {
		this.reactor = reactor;
		
		channel = DatagramChannel.open();
		DatagramSocket s = channel.socket();
		s.setReceiveBufferSize(maxmsgsize);
		s.setSendBufferSize(maxmsgsize);
		
		/*
		if(host == null || host.length() == 0) {
			address = new InetSocketAddress(port);
		} else {
			address = new InetSocketAddress(host, port);
		}
		*/
		s.bind(address);
		channel.configureBlocking(false);
		
		chanhandler = new ChannelHandler() {

			public int getRequestedOps(SelectableChannel _channel) {
				int ret = READ;
				synchronized(messagequeue) {
					if(!messagequeue.isEmpty()) {
						ret |= WRITE;
					}
				}
				return ret;
			}

			public void onReady(int operations, SelectableChannel _channel) {
				if((operations & READ) != 0) {
					/* read */
					ByteBuffer buffer = ByteBuffer.allocate(maxmsgsize);
					buffer.clear();
					try {
						InetSocketAddress address = (InetSocketAddress) channel.receive(buffer);
						if(address != null && msghandler != null) {
							buffer.flip();
							msghandler.onIncoming(new IncomingMessage(buffer, address));
						}
					} catch (IOException e) {
						// ignore, just try again later
					}
				} else if((operations & WRITE) != 0) {
					/* write */
					OutgoingMessage m = null;
					while(true) {
						synchronized(messagequeue) {
							if(messagequeue.isEmpty()) {
								break;
							}					
							m = messagequeue.get(0);
							//System.out.println("Message queue size: " + messagequeue.size());
						}
						ByteBuffer buffer = ByteBuffer.allocate(m.getLength());
						InetSocketAddress address = m.getAddress();
						m.toBuffer(buffer);
						buffer.flip();
						try {
							int nsend = channel.send(buffer, address);
							if(nsend > 0) {
								synchronized(messagequeue) {
									messagequeue.remove(0);
								}
							} else {
								break;
							}
						} catch (IOException e) {
							assert(false);
							e.printStackTrace();
						}
					}
				}
			}
		};
		
		reactor.addSocket(channel, chanhandler);
		
	}
	
	/**
	 * Send message to targethost on port targetport.
	 * 
	 * @param message Message
	 * @param targethost Destination host for message
	 * @param targetport Destination port for message 
	 */
	public void sendMessage(OutgoingMessage message) {
		synchronized(messagequeue) {
			messagequeue.addLast(message);
		}
		reactor.updateChannelOps(channel);
	}
	
	/**
	 * Set the message handler
	 * 
	 * @param handler new message handler
	 */
	public void setHandler(MessageHandler handler) {
		this.msghandler = handler;
	}
	
	/**
	 * Returns the current message handler
	 * 
	 * @return the current message handler
	 */
	public MessageHandler getHandler() {
		return this.msghandler;
	}
	 
	
	private ChannelHandler chanhandler;
	private MessageHandler msghandler = null;
	private NetReactor reactor;
	private DatagramChannel channel;
	
	private final static int maxmsgsize = 65536;

	private LinkedList<OutgoingMessage> messagequeue = new LinkedList<OutgoingMessage>(); 
}
